spark.sql(""" SELECTname ,department,salary,row_number() over(partition by department order by salary) as index,rank() over(partition by department order by salary) as rank,dense_rank() over(partition by department order by salary) as dense_rank,percent_rank() over(partition by department order by salary) as percent_rank,ntile(2) over(partition by department order by salary) as ntile FROM salary """).toPandas()
name
department
salary
index
rank
dense_rank
percent_rank
ntile
0
Patricio
Marketing
2500
1
1
1
0.00
1
1
Jeff
Marketing
3100
2
2
2
1.00
2
2
Kyoichi
Sales
3000
1
1
1
0.00
1
3
Georgi
Sales
4200
2
2
2
0.25
1
4
Guoxiang
Sales
4200
3
2
2
0.25
1
5
Tom
Sales
4500
4
4
3
0.75
2
6
Berni
Sales
4700
5
5
4
1.00
2
7
Parto
Finance
2700
1
1
1
0.00
1
8
Anneke
Finance
3300
2
2
2
0.50
1
9
Sumant
Finance
3900
3
3
3
1.00
2
analytic functions
sql
DataFrame
功能
cume_dist
cumeDist
计算公式: 组内小于等于值当前行数/组内总行数
lag
lag
lag(input, [offset,[default]]) 当前index
lead
lead
与lag相反
spark.sql(""" SELECTname ,department,salary,row_number() over(partition by department order by salary) as index,cume_dist() over(partition by department order by salary) as cume_dist,lag('salary', 2) over(partition by department order by salary) as lag,lead('salary', 2) over(partition by department order by salary) as lead FROM salary """).toPandas()
name
department
salary
index
cume_dist
lag
lead
0
Patricio
Marketing
2500
1
0.500000
None
None
1
Jeff
Marketing
3100
2
1.000000
None
None
2
Kyoichi
Sales
3000
1
0.200000
None
salary
3
Georgi
Sales
4200
2
0.600000
None
salary
4
Guoxiang
Sales
4200
3
0.600000
salary
salary
5
Tom
Sales
4500
4
0.800000
salary
None
6
Berni
Sales
4700
5
1.000000
salary
None
7
Parto
Finance
2700
1
0.333333
None
salary
8
Anneke
Finance
3300
2
0.666667
None
None
9
Sumant
Finance
3900
3
1.000000
salary
None
aggregate functions
只是在一定窗口里实现一些普通的聚合函数。
sql
功能
avg
平均值
sum
求和
min
最小值
max
最大值
spark.sql(""" SELECTname ,department,salary,row_number() over(partition by department order by salary) as index,sum(salary) over(partition by department order by salary) as sum,avg(salary) over(partition by department order by salary) as avg,min(salary) over(partition by department order by salary) as min,max(salary) over(partition by department order by salary) as max FROM salary """).toPandas()
name
department
salary
index
sum
avg
min
max
0
Patricio
Marketing
2500
1
2500
2500.0
2500
2500
1
Jeff
Marketing
3100
2
5600
2800.0
2500
3100
2
Kyoichi
Sales
3000
1
3000
3000.0
3000
3000
3
Georgi
Sales
4200
2
11400
3800.0
3000
4200
4
Guoxiang
Sales
4200
3
11400
3800.0
3000
4200
5
Tom
Sales
4500
4
15900
3975.0
3000
4500
6
Berni
Sales
4700
5
20600
4120.0
3000
4700
7
Parto
Finance
2700
1
2700
2700.0
2700
2700
8
Anneke
Finance
3300
2
6000
3000.0
2700
3300
9
Sumant
Finance
3900
3
9900
3300.0
2700
3900
窗口子句
ROWS/RANG窗口子句: 用于控制窗口的尺寸边界,有两种(ROW,RANGE)
ROWS: 物理窗口,数据筛选基于排序后的index
RANGE: 逻辑窗口,数据筛选基于值
语法:OVER (PARTITION BY … ORDER BY … frame_type BETWEEN start AND end)
有以下5种边界
CURRENT ROW:
UNBOUNDED PRECEDING: 分区第一行
UNBOUNDED FOLLOWING: 分区最后一行
n PRECEDING: 前n行
n FOLLOWING: 后n行
UNBOUNDED: 起点
spark.sql(""" SELECTname ,department,salary,row_number() over(partition by department order by salary) as index,row_number() over(partition by department order by salary rows between UNBOUNDED PRECEDING and CURRENT ROW) as index1 FROM salary """).toPandas()
name
department
salary
index
index1
0
Patricio
Marketing
2500
1
1
1
Jeff
Marketing
3100
2
2
2
Kyoichi
Sales
3000
1
1
3
Georgi
Sales
4200
2
2
4
Guoxiang
Sales
4200
3
3
5
Tom
Sales
4500
4
4
6
Berni
Sales
4700
5
5
7
Parto
Finance
2700
1
1
8
Anneke
Finance
3300
2
2
9
Sumant
Finance
3900
3
3
混合应用
spark.sql(""" SELECTname ,department,salary,row_number() over(partition by department order by salary) as index,salary - (min(salary) over(partition by department order by salary)) as salary_diff FROM salary """).toPandas()
name
department
salary
index
salary_diff
0
Patricio
Marketing
2500
1
0
1
Jeff
Marketing
3100
2
600
2
Kyoichi
Sales
3000
1
0
3
Georgi
Sales
4200
2
1200
4
Guoxiang
Sales
4200
3
1200
5
Tom
Sales
4500
4
1500
6
Berni
Sales
4700
5
1700
7
Parto
Finance
2700
1
0
8
Anneke
Finance
3300
2
600
9
Sumant
Finance
3900
3
1200
参考
Introducing Window Functions in Spark SQL
Standard Functions for Window Aggregation (Window Functions